package com.ruyuan.engine.drools.process;

import com.alibaba.fastjson.JSON;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.advance.RuleBroadcastTable;
import com.ruyuan.engine.pojo.advance.RuleBroadcastTableEnhance;
import com.ruyuan.engine.utils.Constants;
import com.ruyuan.engine.utils.FlinkKafkaUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.util.List;
import java.util.Map;

/**
 * Flink和Drools的整合
 * */
@Slf4j
public class FlinkDrools {

    static MapStateDescriptor<String, KieSession> mapStateDescriptor = new MapStateDescriptor<String, KieSession>("ruleState",String.class,KieSession.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        /**
         * 读取业务数据
         * */
        DataStreamSource<String> source = env.addSource(FlinkKafkaUtils.getKafkaEventSource());
        SingleOutputStreamOperator<ClientLog> clientSource = source.map(m -> JSON.parseObject(m, ClientLog.class));

        /**
         * 规则Topic
         * */
        DataStreamSource<String> ruleSource = env.addSource(FlinkKafkaUtils.getKafkaEventSource());
        BroadcastStream<String> broadcast = ruleSource.broadcast(mapStateDescriptor);
        BroadcastConnectedStream<ClientLog, String> connect = clientSource.connect(broadcast);

        connect.process(new BroadcastProcessFunction<ClientLog, String, String>() {
            @Override
            public void processElement(ClientLog clientLog, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                ReadOnlyBroadcastState<String, KieSession> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                Iterable<Map.Entry<String, KieSession>> entries = broadcastState.immutableEntries();
                for(Map.Entry<String,KieSession> entry:entries ) {
                    KieSession kieSession = entry.getValue();
                    kieSession.insert(clientLog);
                    kieSession.fireAllRules();
                    /**
                     * 规则就要匹配
                     * clientLog：IP是否是 202.200.101.137
                     *      是的话  clientLog.setDeviceType（1） 否则  clientLog.setDeviceType（0）
                     * */
                    out.collect(clientLog.getDeviceType());
                }
            }

            /**
             * 处理规则的数据
             * */
            @Override
            public void processBroadcastElement(String ruleStr, Context ctx, Collector<String> out) throws Exception {
                RuleBroadcastTableEnhance ruleBroadcastTableEnhance = JSON.parseObject(ruleStr, RuleBroadcastTableEnhance.class);
                BroadcastState<String, KieSession> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                List<RuleBroadcastTable> rules = ruleBroadcastTableEnhance.getRules();
                for(RuleBroadcastTable ruleBroadcastTable:rules) {
                    String ruleName = ruleBroadcastTable.getRuleName();
                    String ruleDrl = ruleBroadcastTable.getRuleDrl();
                    KieHelper kieHelper = new KieHelper();
                    kieHelper.addContent(ruleDrl, ResourceType.DRL);
                    KieSession kieSession = kieHelper.build().newKieSession();
                    String operatorRuleType = ruleBroadcastTableEnhance.getOperatorRuleType();
                    if (operatorRuleType.equals(Constants.RULE_ACTIVE)) {
                        log.info("插入规则：{}",ruleName);
                        broadcastState.put(ruleName,kieSession);
                    }else if (operatorRuleType.equals(Constants.RULE_DE_ACTIVE)) {
                        log.info("删除规则：{}",ruleName);
                        broadcastState.remove(ruleName);
                    }else {
                        log.info("无法识别的操作：{}",ruleBroadcastTableEnhance.getOperatorRuleType());
                    }
                }
            }
        }).print();
        env.execute("FlinkDrools");
    }
}
